
/*
 * HadoopReducer.java
 *
 * Created on May 6, 2012, 10:28:19 PM
 */

package edges;


import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
// import org.apache.commons.logging.Log;
// import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.jets3t.service.Constants;

import BasicTypes.Position;

/**
 *
 * @author crazymooner
 */
public class HadoopReducer extends MapReduceBase implements Reducer<Text,Text,IntWritable,IntWritable> {
    // The Karmasphere Studio Workflow Log displays logging from Apache Commons Logging, for example:
    // private static final Log LOG = LogFactory.getLog("edges.HadoopReducer");
	/*
	@Override
    public void reduce(Text key, Iterator<Text> value, OutputCollector<Text, Text> output, Reporter reporter)
            throws IOException {
		Position p = new Position(key.toString());
		Position right = new Position(p.x+1,p.y);
		Position up = new Position(p.x,p.y+1);
		boolean bright = false;
		boolean bup = false;
		boolean bself = false;
		while(value.hasNext()){
			Position temp = new Position(value.next().toString());
			if(temp.equals(p)){
				bself = true;
			}
			if(temp.equals(right)){
				bright = true;
			}
			if(temp.equals(up)){
				bup = true;
			}
		}
		if(bself){
			if(bright){
				output.collect(new Text(p.getString()), new Text(right.getString()));
			}
			if(bup){
				output.collect(new Text(p.getString()), new Text(up.getString()));
			}
		}
    }*/
	
	@Override
    public void reduce(Text key, Iterator<Text> value, OutputCollector<IntWritable, IntWritable> output, Reporter reporter)
            throws IOException {
		Position p = new Position(key.toString());
		Position right = new Position(p.x+1,p.y);
		Position up = new Position(p.x,p.y+1);
		Position upright = new Position(p.x + 1, p.y + 1);
		Position upleft = new Position(p.x - 1, p.y + 1);
		boolean bright = false;
		boolean bup = false;
		boolean bself = false;
		boolean bupright = false;
		boolean bupleft = false;
		while(value.hasNext()){
			Position temp = new Position(value.next().toString());
			if(temp.equals(p)){
				bself = true;
				continue;
			}
			if(temp.equals(right)){
				bright = true;
				continue;
			}
			if(temp.equals(up)){
				bup = true;
				continue;
			}
			if(temp.equals(upleft)){
				bupleft = true;
				continue;
			}
			if(temp.equals(upright)){
				bupright = true;
				continue;
			}
		}
		if(bself){
			if(bright){
				output.collect(new IntWritable(p.toNumber()), new IntWritable(right.toNumber()));
				//output.collect(new IntWritable(p.toNumber()), new IntWritable(BasicTypes.Constants.SIZE));
			}
			if(bup){
				output.collect(new IntWritable(p.toNumber()), new IntWritable(up.toNumber()));
			}
			if(bupleft){
				output.collect(new IntWritable(p.toNumber()), new IntWritable(upleft.toNumber()));
			}
			if(bupright){
				output.collect(new IntWritable(p.toNumber()), new IntWritable(upright.toNumber()));
			}
		}

    }
	
}
